#ifndef ASYNC_HPP
#define ASYNC_HPP

#include "sink.hpp"
#include <thread>
#include <atomic>
#include <vector>
#include <condition_variable>

namespace ly
{
    class Buffer
    {
    public:
        Buffer() : _pos_read(0), _pos_write(0) { _buffer.reserve(1024); }
        void add_message(const std::pair<std::shared_ptr<std::string>, std::shared_ptr<std::vector<Sink::ptr>>> &message)
        {
            if(_pos_write == _buffer.size()) _buffer.emplace_back(message);
            else _buffer[_pos_write] = message;
            ++_pos_write;
        }

        std::pair<std::shared_ptr<std::string>, std::shared_ptr<std::vector<Sink::ptr>>> get_message()
        {
            assert(!empty());
            return _buffer[_pos_read++];
        }

        bool empty() { return _pos_read == _pos_write; }

        void clear()
        {
            _pos_read = _pos_write = 0;
        }

        int size() { return _pos_write - _pos_read; }

    private:
        size_t _pos_read, _pos_write;
        std::vector<std::pair<std::shared_ptr<std::string>, std::shared_ptr<std::vector<Sink::ptr>>>> _buffer;
    };

    class AsyncLooper
    {
    public:
        static AsyncLooper &getInstance()
        {
            static AsyncLooper looper;
            return looper;
        }

        void add_message(const std::pair<std::shared_ptr<std::string>, std::shared_ptr<std::vector<Sink::ptr>>> &message)
        {
            {
                std::lock_guard<std::mutex> lock(_mutex);
                _buffer[_write].add_message(message);
            }
            _con.notify_one();
        }

        void stop()
        {
            _stop = true;
            _con.notify_one();
        }

        ~AsyncLooper()
        {
            stop();
            if (_consume.joinable())
                _consume.join();
        }

    private:
        AsyncLooper() : _stop(false), _buffer(2), _read(0), _write(1)
        {
            _consume = std::thread(&AsyncLooper::consume, this);
        }
        AsyncLooper(const AsyncLooper&) = delete;
        void consume()
        {
            while (true)
            {
                while (_buffer[_read].empty() == false)
                {
                    auto message = _buffer[_read].get_message();
                    for (auto &sink : *message.second)
                        sink->log(message.first->c_str(), message.first->size());
                }

                _buffer[_read].clear();
                std::unique_lock<std::mutex> lock(_mutex);

                _con.wait(lock, [this] { return _buffer[_write].size() || _stop; });

                if (_buffer[_write].empty() && _stop)
                    break;
                std::swap(_read, _write);
            }
        }

    private:
        std::atomic<bool> _stop;
        std::mutex _mutex;
        std::thread _consume;
        std::condition_variable _con;
        std::vector<Buffer> _buffer;
        size_t _read, _write;
    };
} // namespace ly

#endif // #define ASYNC_HPP